package com.feidee.fd.sml.algorithm.component.validation

import com.feidee.fd.sml.algorithm.component.{AbstractComponent, BasicParam}
import com.feidee.fd.sml.algorithm.forecast.SparkForecast
import org.apache.spark.SparkException

/**
  * @Author songhaicheng
  * @Date 2019/4/24 15:59
  * @Description
  * @Reviewer
  */
class ValidationComponent extends AbstractComponent[ValidationParam] with Serializable {

  override def apply(paramStr: String): Unit = {
    logInfo("parsing parameter")
    val param = parseParam(paramStr)
    logInfo("verifying parameter")
    param.verify()

    // 得到 Spark 模型跑测试数据的结果
    val testingData = spark.sql(param.testingSql)
    val forecast = new SparkForecast(spark,null, param.sparkModels: _*)
    val result = forecast.predict(testingData)

    if (param.assertions.forall(ass => {
      if (!ass.apply(result, ass.className, ass.params, param)) {
        throw new SparkException("校验失败")
      } else {
        true
      }
    })) {
      param.actions.forall(act => {
        if (!act.apply(result, act.className, act.params, param)) {
          throw new SparkException("执行失败")
        } else {
          true
        }
      })
    }

    // 输出预测结果
    outputData(result, param)
  }

}

case class ValidationParam(
                            override val input_pt: String,
                            override val output_pt: String,
                            override val hive_table: String,
                            override val flow_time: String,
                            sparkModels: Array[String],
                            testingSql: String,
                            assertions: Array[Morpher],
                            actions: Array[Morpher],
                            env: String,
                            project_name: String,
                            submitter: String
                          ) extends BasicParam {

  def this() = this(null, null, null, null, Array.empty[String], null, Array.empty[Morpher], Array.empty[Morpher],
    null, null, "anonymous")

  override def verify(): Unit = {
    super.verify()
    require(tool.isNotNull(testingSql), "必须指定 testingSql 测试数据")
    require(sparkModels.length > 0, "必须指定 sparkModels 模型路径")
  }

  override def toMap: Map[String, Any] = {
    var map = super.toMap
    map += ("sparkModels" -> sparkModels)
    map += ("testingSql" -> testingSql)
    map += ("assertions" -> assertions)
    map += ("env" -> env)
    map += ("project_name" -> project_name)
    map += ("submitter" -> submitter)
    map
  }

}

object ValidationComponent {
  def apply(paramStr: String): Unit = {
    new ValidationComponent()(paramStr)
  }

  def main(args: Array[String]): Unit = {
    ValidationComponent(args(0))
  }
}
